package spring.cloud.tasks.executor.executor.task;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import spring.cloud.tasks.common.curator.Curator;
import spring.cloud.tasks.common.curator.CuratorFactory;
import spring.cloud.tasks.common.curator.CuratorInfo;
import spring.cloud.tasks.common.nodepaths.ExecutorTaskNodePath;
import spring.cloud.tasks.common.nodepaths.TaskNodePaths;
import spring.cloud.tasks.executor.executor.Executor;
import spring.cloud.tasks.executor.executor.TaskExecutorInitAndTaskItemAssignListenerService;
import spring.cloud.tasks.executor.executor.task.task_configuration.TaskConfiguration;
import spring.cloud.tasks.executor.executor.task.task_item_execute.TaskItemExecuteRenewalInfoService;
import spring.cloud.tasks.executor.executor.task.task_trigger.TaskExecutorInfoService;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 接受是否该执行器是否允许运行任务,如果允许则到了时间执行。执行完毕后。如果该任务没有分配到上面则自己释放自己。如果允许任务跑在上面则创建任务执行器。
 */
@Slf4j
@Setter
@Getter
public class TaskExecutor {
    private final Curator curator;

    private final TaskExecutorNodeCacheOrTreeCacheManager taskExecutorNodeCacheOrTreeCacheManager;
    //
    private final TaskConfiguration taskConfiguration;

    private final TaskListenerHolderManager taskListenerHolderManager;
    private final TaskExecutorInfoService taskExecutorInfoService;
    private final TaskItemExecuteRenewalInfoService taskItemExecuteInfoService;
    public volatile boolean needStop = false;//外部尝试关闭任务执行器
    public ReentrantLock lock = new ReentrantLock();
    private String executorId;
    private Executor executor;

    private TaskExecutorInitAndTaskItemAssignListenerService taskInitService;
    private String taskId;

    private TaskExecuteEngine taskExecuteEngine;
    private AtomicBoolean shutdown = new AtomicBoolean(false);
    private ScheduledExecutorService taskExecuteInfoCollectScheduledExecutorService;

    public TaskExecutor(Executor executor, final Curator nameSpaceDataManager, final TaskConfiguration taskConfiguration, TaskExecutorInitAndTaskItemAssignListenerService taskInitService) {
        this.curator = nameSpaceDataManager;
        this.executorId = executor.getExecutorId();
        this.executor = executor;
        this.taskInitService = taskInitService;
        this.taskId = taskConfiguration.getTaskId();
        this.taskConfiguration = taskConfiguration;

        this.taskExecutorNodeCacheOrTreeCacheManager = new TaskExecutorNodeCacheOrTreeCacheManager(nameSpaceDataManager.getCuratorFramework(), executorId, taskId);
        //


        this.taskExecutorInfoService = new TaskExecutorInfoService(this);
        this.taskItemExecuteInfoService = new TaskItemExecuteRenewalInfoService(this);
        //

        this.taskListenerHolderManager = new TaskListenerHolderManager(this);
        executor.taskIdToTaskExecutorMap.addTaskExecutor(taskId, this);
    }

    public void needStop() {
        this.needStop = true;
        if (taskExecuteEngine != null) {
            taskExecuteEngine.needStop = true;
        }
    }

    /**
     * 初始化任务.
     */
    public void init() {
        try {
            startAll();
            createTaskExecuteEngine();
            taskExecutorInfoService.persistExecutorOnline();
        } catch (Throwable t) {
            shutdown();
            throw t;
        }
    }

    private void startAll() {
        taskListenerHolderManager.start();
        //


        taskExecutorInfoService.start();
        taskItemExecuteInfoService.start();
        //

        taskExecutorInfoService.resetCount();


    }

    private void createTaskExecuteEngine() {
        taskExecuteEngine = new TaskExecuteEngine();
        taskExecuteEngine.setNameSpace(executor.getNamespaceKey());
        taskExecuteEngine.setExecutorId(executorId);
        taskExecuteEngine.setTaskId(taskId);
        taskExecuteEngine.setTaskExecutor(this);
        //

        taskExecuteEngine.setTaskItemExecuteInfoService(taskItemExecuteInfoService);
        //

        initTaskExecuteInfoCollectScheduledExecutorService();
    }

    /**
     * 立刻启动任务.
     */
    public void rightNowExecute() {

        //taskExecuteEngine.getClicker().rightNowExecute();
        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {
                taskExecuteEngine.startTaskItemExecuteThread();
            }
        });
        thread.start();
    }

    public void shutdown() {
        synchronized (shutdown) {
            shutdown.set(true);

            // 关闭Listener
            taskListenerHolderManager.stop();

            // 关闭任务：关闭调度器，强杀业务
            stopTaskExecuteInfoCollectScheduledExecutorService();

            // 关闭服务


            taskExecutorInfoService.stop();


            // 关闭TreeCache
            taskExecutorNodeCacheOrTreeCacheManager.shutdown();
            deleteTaskExecutorNode();

            // 移除任务注册表
            this.executor.taskIdToTaskExecutorMap.removeTaskExecutor(taskId);
            this.executor.taskIdToTaskExecutorMap.taskIdSet.remove(taskId);
        }
    }

    /**
     * 删除ZK结点
     */
    public void deleteTaskExecutorNode() {
        CuratorInfo curatorInfo = (curator).getCuratorInfo();
        Curator curator = CuratorFactory.getCurator(curatorInfo, CuratorFactory.shortTime_true);
        try {

            curator.delete(ExecutorTaskNodePath.getExecutorTaskNode(executorId, taskId));//任务分配
            curator.delete(TaskNodePaths.getTaskExecutorNodePath(taskId, executorId));//只是标记
        } catch (Throwable t) {
            log.error("删除任务执行节点失败:" + t.getMessage(), t);
        } finally {
            curator.close();
        }
    }

    /**
     * 调度器分配的任务
     */


    public void initTaskExecuteInfoCollectScheduledExecutorService() {
        if (taskExecuteInfoCollectScheduledExecutorService != null) {
            taskExecuteInfoCollectScheduledExecutorService.shutdown();
        }
        taskExecuteInfoCollectScheduledExecutorService = getTaskExecuteInfoCollectScheduledExecutorService(this);
        TaskStateMaintenanceThread taskStateMaintenanceThread = new TaskStateMaintenanceThread(this);
        taskExecuteInfoCollectScheduledExecutorService.scheduleAtFixedRate(taskStateMaintenanceThread, 10, 10, TimeUnit.SECONDS);
    }

    public void stopTaskExecuteInfoCollectScheduledExecutorService() {
        if (taskExecuteInfoCollectScheduledExecutorService != null) {
            taskExecuteInfoCollectScheduledExecutorService.shutdown();
        }
    }

    public ScheduledExecutorService getTaskExecuteInfoCollectScheduledExecutorService(final TaskExecutor taskExecutor) {
        ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {//
            @Override
            public Thread newThread(Runnable runnable) {
                Thread thread = new Thread(runnable, taskExecutor.getExecutorId() + "_" + taskExecutor.getTaskId() + TaskStateMaintenanceThread.class.getSimpleName());
                thread.setDaemon(false);
                thread.setPriority(Thread.NORM_PRIORITY);
                return thread;
            }
        });
        return scheduledExecutorService;
    }
}
